package raft

import (
	`time`
)

type logEntry struct {
	//LogContent *logContent
	LogContent interface{}
	Index      int
	Term       int
}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PreLogIndex  int
	PreLogTerm   int
	Entries      []logEntry
	LeaderCommit int
}

type AppendEntriesReply struct {
	Term        int
	CommitIndex int
	Success     bool
}

func (rf *Raft) getCommitIndex(isLock ...bool) int {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLeaderState.mu.Lock()
		defer rf.raftLeaderState.mu.Unlock()
	}
	return rf.commitIndex
}

func (rf *Raft) getNextIndex(server int) int {
	rf.raftLeaderState.mu.Lock()
	defer rf.raftLeaderState.mu.Unlock()
	return rf.nextIndex[server]
}

func (rf *Raft) setNextIndex(server int, index int) {
	rf.raftLeaderState.mu.Lock()
	defer rf.raftLeaderState.mu.Unlock()
	rf.nextIndex[server] = index
}

// set successfully return true, other return false
func (rf *Raft) setIsSyncIfNot(server int) bool {
	rf.raftLeaderState.mu.Lock()
	defer rf.raftLeaderState.mu.Unlock()
	if !rf.isSynchronizing[server] {
		rf.isSynchronizing[server] = true
		return true
	}
	return false
}

func (rf *Raft) setNotSync(server int, wait time.Duration) {
	time.Sleep(wait)
	rf.raftLeaderState.mu.Lock()
	defer rf.raftLeaderState.mu.Unlock()
	rf.isSynchronizing[server] = false
}

func (rf *Raft) getEmptyLog() logEntry {
	return logEntry{nil, 0, 0}
}

func (rf *Raft) lastLog(isLock ...bool) logEntry {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}

	if len(rf.raftLogState.logs) == 0 {
		return rf.getEmptyLog()
	}
	return rf.raftLogState.logs[len(rf.raftLogState.logs)-1]
}

// start with 1
func (rf *Raft) getLog(i int, isLock ...bool) logEntry {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}

	if i <= 0 || i > len(rf.logs) {
		return rf.getEmptyLog()
	}
	return rf.logs[i-1]
}

// start with 1; if n == 0, return []
func (rf *Raft) getLogsFrom(n int, isLock ...bool) []logEntry {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}

	if n <= 0 || len(rf.logs) == 0 || n > len(rf.logs) {
		return []logEntry{}
	}
	return rf.logs[n-1:]
}

func (rf *Raft) getLogs(n, m int, isLock ...bool) []logEntry {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}

	if n <= 0 || len(rf.logs) == 0 || n > len(rf.logs) {
		return []logEntry{}
	}
	return rf.logs[n-1 : m]
}

// return log index
func (rf *Raft) makeAppendLog(command interface{}) int {
	rf.raftLogState.mu.Lock()
	defer rf.raftLogState.mu.Unlock()
	index := len(rf.logs) + 1
	log := logEntry{command, index, rf.getCurrentTeam()}
	rf.logs = append(rf.logs, log)
	rf.persist(false)
	return index
}

func (rf *Raft) nowLogIndex(isLock ...bool) int {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}

	return len(rf.logs)
}

func (rf *Raft) setCommitIndex(index int) {
	rf.raftLogState.mu.Lock()
	defer rf.raftLogState.mu.Unlock()
	rf.commitIndex = index
}

func (rf *Raft) alwaysBroadcastEntries() {
	for true {
		time.Sleep(BroadcastInterval)
		if rf.killed() || !rf.isLeader() {
			continue
		}
		FPrintf("[%d] is leader, will broadcastEntries", rf.me)
		go rf.broadcastEntries()
	}
}

func (rf *Raft) broadcastEntries() {
	defer NeverPanic()

	getSupport := make(chan struct{})
	defer close(getSupport)

	commitIndex, nowLogIndex, hasMyTermLog := func() (int, int, bool) {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
		return rf.commitIndex, rf.nowLogIndex(false), rf.lastLog(false).Term == rf.currentTeam
	}()
	go rf.doBroadcastEntries(getSupport, nowLogIndex)

	if commitIndex == nowLogIndex || !hasMyTermLog { // 并没有新的东西被提交了 || 最新的一条记录不是自己的，不能commit
		return
	}

	tick := time.NewTicker(HeartbeatTimeout)
	defer tick.Stop()

	count, majority := 1, len(rf.peers)/2+1
	for true {
		//DPrintf("[%d] in election, count: %d, majority: %d, isPreVote: %t", rf.me, count, majority, isPreVote)
		select {
		case <-getSupport:
			count++
			DPrintf("[%d] get vote support", rf.me)
			if count >= majority {
				rf.commitEntries(nowLogIndex)
				return
			}
		case <-tick.C:
			//DPrintf("[%d] max vote for time, goroutine out. term: %d", rf.me, args.Term)
			return
		}
	}
}

func (rf *Raft) doBroadcastEntries(c chan<- struct{}, end int) {
	rf.raftLeaderState.mu.Lock()
	defer rf.raftLeaderState.mu.Unlock()
	DPrintf("[%d] will doBroadcastEntries, my log: %d", rf.me, rf.logs)
	for i := range rf.peers {
		if i == rf.me {
			continue
		}

		go func(server, start, end int) {
			defer NeverPanic()
			ok, reply := rf.translateEntries(server, start, end, false)

			if ok && reply != nil {
				if reply.Success {
					DPrintf("[%d] sync server %d success without control", rf.me, server)
					rf.setNextIndex(server, end+1)
					c <- struct{}{}
					return
				}

				rf.setNextIndex(server, reply.CommitIndex+1)
				if ok, reply = rf.translateEntries(server, reply.CommitIndex+1, end, true); ok && reply != nil && reply.Success {
					rf.setNextIndex(server, end+1)
					c <- struct{}{}
				}
			}
		}(i, rf.nextIndex[i], end)
	}
}

func (rf *Raft) translateEntries(server, start, end int, control bool) (bool, *AppendEntriesReply) {
	if control {
		if !rf.setIsSyncIfNot(server) {
			return false, nil
		}
		go rf.setNotSync(server, HeartbeatTimeout)
	}

	DPrintf("[%d] sync server %d", rf.me, server)
	args, reply := rf.makeAppendEntriesArgsReply(start, end)
	DPrintf("[%d] sync server %d, args: %d", rf.me, server, args)

	if !rf.isLeader() {
		DPrintf("[%d] sync server %d stop, I am not Leader any more", rf.me, server)
		return false, nil
	}
	ok := rf.sendAppendEntries(server, &args, &reply)
	DPrintf("[%d] sync server %d success: %t", rf.me, server, reply.Success)
	return ok, &reply
}

// start with 1
func (rf *Raft) makeAppendEntriesArgsReply(start, end int, isLock ...bool) (AppendEntriesArgs, AppendEntriesReply) {
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()

	}
	entries := rf.getLogs(start, end, false)
	preLog := rf.getLog(start-1, false)
	args := AppendEntriesArgs{Term: rf.currentTeam, LeaderId: rf.me, PreLogIndex: preLog.Index, PreLogTerm: preLog.Term,
		Entries: entries, LeaderCommit: rf.commitIndex}
	reply := AppendEntriesReply{}
	return args, reply
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	reply.Success = false
	reply.Term = rf.getCurrentTeam()
	reply.CommitIndex = rf.getCommitIndex()
	if args.Term < reply.Term { // 比自己期次小
		return
	}

	FPrintf("[%d] get heartbeat from %d, update timeout", rf.me, args.LeaderId)
	rf.updateTimeOut()
	rf.setCurrentTeam(args.Term)
	rf.becomeFollower() // 到这里为止，表示认可了心跳，带不一定认可log

	DPrintf("[%d] get args preLogIndex: %d, preLogTerm: %d, my commitIndex: %d", rf.me, args.PreLogIndex, args.PreLogTerm, reply.CommitIndex)
	//if len(args.Entries) == 0 {
	//	DPrintf("[%d] I am Follower, get HeartBeat", rf.me)
	//	return
	//}
	if rf.nowLogIndex() < args.PreLogIndex { // leader指定的前一条日志不存在
		DPrintf("[%d] ！！！the pre log not exist, my log: %d, pre-index: %d", rf.me, rf.logs, args.PreLogIndex)
		return
	}

	if args.PreLogTerm != rf.getLog(args.PreLogIndex).Term { // 期次不一致
		DPrintf("[%d] ！！！the pre log term not same, my log: %d, ", rf.me, rf.logs)
		return
	}

	reply.Success = true
	rf.appendCommitEntries(args.PreLogIndex, args.Entries, args.LeaderCommit)
}

func (rf *Raft) appendCommitEntries(index int, entries []logEntry, leaderCommit int) {
	rf.raftLogState.mu.Lock()
	defer rf.raftLogState.mu.Unlock()
	rf.appendEntries(index, entries, false)
	rf.commitEntries(leaderCommit, false)
}

// start with 1
func (rf *Raft) appendEntries(index int, entries []logEntry, isLock ...bool) {
	DPrintf("[%d] as follower, will append entries: %d", rf.me, entries)
	if len(entries) == 0 {
		return
	}
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}
	rf.logs = append(rf.logs[:index], entries...)
	rf.persist(false)

}

// start with 1
func (rf *Raft) commitEntries(leaderCommit int, isLock ...bool) {
	DPrintf("[%d] as follower, will commit entries: %d, my logs: %d", rf.me, leaderCommit, 1)
	if len(isLock) == 0 || isLock[0] {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}
	rf.commitIndex = leaderCommit
}

func (rf *Raft) alwaysCommitLogs() {
	commitIndex := 0
	for {
		time.Sleep(CommitInterval)
		nextCommitIndex := rf.commitIndex
		rf.sendApplyMsg(commitIndex, nextCommitIndex)
		commitIndex = nextCommitIndex
	}
}

func (rf *Raft) sendApplyMsg(commitIndex int, leaderCommit int) {
	DPrintf("[%d] will send apply msg, my log: %d, commitIndex: %d, leaderCommit %d", rf.me, 1, commitIndex, leaderCommit)
	for i := commitIndex; i < len(rf.logs) && i < leaderCommit; i++ {
		applyMsg := ApplyMsg{true, rf.logs[i].LogContent, rf.logs[i].Index}
		DPrintf("[%d] send apply msg: %d, %d", rf.me, applyMsg.Command, applyMsg.CommandIndex)
		rf.applyCh <- applyMsg
		DPrintf("[%d] finish send apply msg: %d, %d", rf.me, applyMsg.Command, applyMsg.CommandIndex)
	}
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}
